﻿


using IotContract.Interfaces;

using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using MQTTnet.Server;

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;

using System.Threading;

using IotContract.Interfaces.Process;
using Microsoft.Extensions.DependencyInjection;
using IotContract.Models.Mqtt;

using XT.Common.Services;
using XT.Common.Config;
using XT.Common.Extensions;

namespace IotServerService.Services.Mqtts;
/// <summary>
/// 添加Mqtt服务，对外发布协议数据 
/// </summary>
public class MqttManager
{ 
    private string ServerClientId;
    private string _userName;
    private string _password;
    private readonly ILogService _logService;
    private MqttServer _mqttServer;
    private CancellationTokenSource _cancellationTokenSource;
    private MqttConfig _mqttConfig;
    private Dictionary<string,int> subscribedTopics = new Dictionary<string, int>();//订阅主题发布未消费计数
    public  MqttManager (ILogService logService)
    {
        _logService = logService;
        _mqttConfig = AppSettings.GetObjData<MqttConfig>();
    }

    public MqttServer GetMqttServer()
    {
        return _mqttServer;        
    }
    /// <summary>
    /// Mqtt启动
    /// </summary>
    /// <param name="cfg">Mqtt配置类</param>
    /// <returns></returns>
    public async Task<bool> InitAsync()
    {
        ServerClientId = _mqttConfig.ServerClientId;
        _userName = _mqttConfig.UserName;
        _password = _mqttConfig.Password;
        MqttServerOptionsBuilder optionsBuilder = new MqttServerOptionsBuilder();
        optionsBuilder.WithDefaultEndpoint();
        optionsBuilder.WithDefaultEndpointPort(_mqttConfig.Port); // 设置 服务端 端口号
        optionsBuilder.WithConnectionBacklog(_mqttConfig.LinkMax); // 最大连接数
        optionsBuilder.WithMaxPendingMessagesPerClient(_mqttConfig.maxPendingMessagesPerClient);
        MqttServerOptions options = optionsBuilder.Build();
        _mqttServer = new MqttFactory().CreateMqttServer(options);
        _mqttServer.ClientConnectedAsync -= _mqttServer_ClientConnectedAsync; //客户端连接事件
        _mqttServer.ClientConnectedAsync += _mqttServer_ClientConnectedAsync; //客户端连接事件
        _mqttServer.ClientDisconnectedAsync -= _mqttServer_ClientDisconnectedAsync; // 客户端关闭事件
        _mqttServer.ClientDisconnectedAsync += _mqttServer_ClientDisconnectedAsync; // 客户端关闭事件
        _mqttServer.ApplicationMessageNotConsumedAsync -= _mqttServer_ApplicationMessageNotConsumedAsync; // 消息未接收事件
        _mqttServer.ApplicationMessageNotConsumedAsync += _mqttServer_ApplicationMessageNotConsumedAsync; // 消息未接收事件
        _mqttServer.ClientSubscribedTopicAsync -= _mqttServer_ClientSubscribedTopicAsync; // 客户端订阅主题事件
        _mqttServer.ClientSubscribedTopicAsync += _mqttServer_ClientSubscribedTopicAsync; // 客户端订阅主题事件
        _mqttServer.ClientUnsubscribedTopicAsync -= _mqttServer_ClientUnsubscribedTopicAsync; // 客户端取消订阅事件
        _mqttServer.ClientUnsubscribedTopicAsync += _mqttServer_ClientUnsubscribedTopicAsync; // 客户端取消订阅事件
        _mqttServer.StartedAsync -= _mqttServer_StartedAsync; // 启动后事件
        _mqttServer.StartedAsync += _mqttServer_StartedAsync; // 启动后事件
        _mqttServer.StoppedAsync -= _mqttServer_StoppedAsync; // 关闭后事件
        _mqttServer.StoppedAsync += _mqttServer_StoppedAsync; // 关闭后事件
        _mqttServer.InterceptingPublishAsync -= _mqttServer_InterceptingPublishAsync; // 消息接收事件
        _mqttServer.InterceptingPublishAsync += _mqttServer_InterceptingPublishAsync; // 消息接收事件
        _mqttServer.ValidatingConnectionAsync -= _mqttServer_ValidatingConnectionAsync; // 用户名和密码验证有关
        _mqttServer.ValidatingConnectionAsync += _mqttServer_ValidatingConnectionAsync; // 用户名和密码验证有关

        return true;
    }
    public async Task<bool> StartAsync()
    {
        bool res = false;
        try
        {
            await _mqttServer.StartAsync();
            res = true;
        }
        catch (Exception ex)
        {
            _logService.Log("MQTT启动异常：" + ex.StackTrace);
            res = false;
        }
        return res;
    }
    /// <summary>
    /// 客户端订阅主题事件
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    private Task _mqttServer_ClientSubscribedTopicAsync(ClientSubscribedTopicEventArgs arg)
    {
        _logService.Log($"Mqtt客户端{arg.ClientId}订阅的主题{arg.TopicFilter}");
        if(!subscribedTopics.Keys.Contains(arg.TopicFilter.Topic))
        {
            subscribedTopics.Add(arg.TopicFilter.Topic,0);
        }
        else
        {
            subscribedTopics[arg.TopicFilter.Topic] =  0;
        }
        return Task.CompletedTask;
    }

    /// <summary>
    /// 关闭后事件
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    private Task _mqttServer_StoppedAsync(EventArgs arg)
    {
        //Console.WriteLine($"StoppedAsync：MQTT服务已关闭……");
        _logService.Log($"StoppedAsync：MQTT服务已关闭……");
        return Task.CompletedTask;
    }

    /// <summary>
    /// 用户名和密码验证有关
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    private Task _mqttServer_ValidatingConnectionAsync(ValidatingConnectionEventArgs arg)
    {
        arg.ReasonCode = MqttConnectReasonCode.Success;
        if ((arg.UserName ?? string.Empty) != _userName || (arg.Password ?? String.Empty) != _password)
        {
            arg.ReasonCode = MqttConnectReasonCode.Banned;
            //Console.WriteLine($"ValidatingConnectionAsync：客户端ID=【{arg.ClientId}】用户名或密码验证错误 ");
            _logService.Log($"ValidatingConnectionAsync：客户端ID=【{arg.ClientId}】用户名或密码验证错误 ");
        }
        return Task.CompletedTask;
    }

    /// <summary>
    /// 消息接收事件
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    private Task _mqttServer_InterceptingPublishAsync(InterceptingPublishEventArgs arg)
    {
        if (string.Equals(arg.ClientId, ServerClientId))
        {
            return Task.CompletedTask;
        }
        Task<IList<MqttClientStatus>> s = _mqttServer.GetClientsAsync();
        var s1 = _mqttServer.GetRetainedMessagesAsync();
        var s2 = _mqttServer.GetSessionsAsync();
        
        _logService.Log($"InterceptingPublishAsync：客户端ID=【{arg.ClientId}】 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
        return Task.CompletedTask;

    }

    /// <summary>
    /// 启动后事件
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    private Task _mqttServer_StartedAsync(EventArgs arg)
    {
        _logService.Log($"StartedAsync：MQTT服务已启动……");
        return Task.CompletedTask;
    }

    /// <summary>
    /// 客户端取消订阅事件
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    private Task _mqttServer_ClientUnsubscribedTopicAsync(ClientUnsubscribedTopicEventArgs arg)
    {
        _logService.Log($"ClientUnsubscribedTopicAsync：客户端ID=【{arg.ClientId}】已取消订阅的主题=【{arg.TopicFilter}】  ");
        return Task.CompletedTask;
    }

    /// <summary>
    /// 消息接收事件
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    private Task _mqttServer_ApplicationMessageNotConsumedAsync(ApplicationMessageNotConsumedEventArgs arg)
    {
        //Console.WriteLine($"ApplicationMessageNotConsumedAsync：发送端ID=【{arg.SenderId}】 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【{Encoding.UTF8.GetString(arg.ApplicationMessage.Payload)}】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
        _logService.Log($"ApplicationMessageNotConsumedAsync：发送端ID=【{arg.SenderId}】 Topic主题=【{arg.ApplicationMessage.Topic}】 消息=【】 qos等级=【{arg.ApplicationMessage.QualityOfServiceLevel}】");
        if (!subscribedTopics.Keys.Contains(arg.ApplicationMessage.Topic))
        {
            subscribedTopics.Add(arg.ApplicationMessage.Topic, 0);
        }
        subscribedTopics[arg.ApplicationMessage.Topic]++;
        return Task.CompletedTask;
    }

    /// <summary>
    /// 客户端断开时候触发
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    /// <exception cref="NotImplementedException"></exception>
    private Task _mqttServer_ClientDisconnectedAsync(ClientDisconnectedEventArgs arg)
    {
        //Console.WriteLine($"ClientDisconnectedAsync：客户端ID=【{arg.ClientId}】已断开, 地址=【{arg.Endpoint}】  ");
        _logService.Log($"ClientDisconnectedAsync：客户端ID=【{arg.ClientId}】已断开, 地址=【{arg.Endpoint}】  ");
        return Task.CompletedTask;
    }

    /// <summary>
    /// 客户端连接时候触发
    /// </summary>
    /// <param name="arg"></param>
    /// <returns></returns>
    private Task _mqttServer_ClientConnectedAsync(ClientConnectedEventArgs arg)
    {
        
        //Console.WriteLine($"ClientConnectedAsync：客户端ID=【{arg.ClientId}】已连接, 用户名=【{arg.UserName}】地址=【{arg.Endpoint}】  ");
        _logService.Log($"ClientConnectedAsync：客户端ID=【{arg.ClientId}】已连接, 用户名=【{arg.UserName}】地址=【{arg.Endpoint}】  ");
        var s =_mqttServer.ServerSessionItems;
        return Task.CompletedTask;
    }

    public async Task<bool> StopAsync()
    {
        await _mqttServer.StopAsync();
        if (!_mqttServer.IsStarted)
        {
            return true;
        }
        return false;
    }
    /// <summary>
    /// mqtt发布
    /// </summary>
    /// <param name="topic">发布主题</param>
    /// <param name="data">发布内容</param>
    /// <param name="QoS">服务质量: 
    /// （QoS）:  0,AtMostOnce 最多一次，接收者不确认收到消息，并且消息不被发送者存储和重新发送提供与底层 TCP 协议相同的保证。
    /// 1,AtLeastOnce: 保证一条消息至少有一次会传递给接收方。发送方存储消息，直到它从接收方收到确认收到消息的数据包。一条消息可以多次发送或传递。
    /// 2,ExactlyOnce: 保证每条消息仅由预期的收件人接收一次。级别2是最安全和最慢的服务质量级别，保证由发送方和接收方之间的至少两个请求/响应（四次握手）。
    /// </param>
    /// <param name="isRetain">服务端是否保留消息,如果有新的订阅者连接，就会立马收到该消息</param>
    /// <param name="senderClientId">服务质量</param>
    public async Task<bool> PublishData(string topic, string data, int QoS, bool isRetain, string? senderClientId = "")
    {
        if (topic.IsNullOrEmpty() || data.IsNullOrEmpty()) 
        {
            return false;  
        }
        //主题未被订阅，或已订阅但累计5次发布未被消费，停止发送；需客户端重新订阅才会发送
        if(!subscribedTopics.Keys.Contains(topic) || (subscribedTopics.Keys.Contains(topic) && subscribedTopics[topic] > 5))
        {
            return false;
        }
        try
        {

            var message = new MqttApplicationMessage()
            {
                Topic = topic,
                PayloadSegment = Encoding.Default.GetBytes(data),
                QualityOfServiceLevel = (MqttQualityOfServiceLevel)QoS,
                Retain = isRetain
            };
            
            if (_mqttServer.IsNullOrEmpty() || !_mqttServer.IsStarted)
            {
                return false;
            }
            await _mqttServer.GetClientsAsync();
            try
            {
                await _mqttServer.InjectApplicationMessage(new InjectedMqttApplicationMessage(message) // 发送消息给有订阅主题的客户端
                {
                    SenderClientId = String.IsNullOrEmpty(senderClientId) ? "Server" : senderClientId
                });
            }
            catch (Exception e)
            {
                _logService.Log($"{DateTime.Now}发布处理异常2{e.Message}");
            }
        }
        catch (Exception e)
        {
            _logService.Log($"{DateTime.Now}发布处理异常{e.Message}");
        }
        
        return true;
    }


    public async Task<bool> MqttStop()
    {
        _logService.Log($"Mqtt停止运行");
        _cancellationTokenSource?.Cancel();
        return true;
    }



}
